package com.huawei.hbase.data;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.MultipleColumnPrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;

public class ScanData {
    public static void main(String[] args) throws IOException {
        //??hbase
        Configuration conf = HBaseConfiguration.create();
        //???zookeeper??????????zookeeper??
        conf.set("hbase.zookeeper.quorum",
                "xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx,xxx.xxx.xxx.xxx");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        Connection connection = ConnectionFactory.createConnection(conf);

        Table hTable = connection.getTable(TableName.valueOf("users"));


        Scan scan = new Scan();
        //????rowkey
        scan.withStartRow(Bytes.toBytes("row1"));
        scan.withStopRow(Bytes.toBytes("row5"));

        //????filter
        FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        byte[][] prefixes = new byte[2][];
        prefixes[0] = Bytes.toBytes("id");
        prefixes[1] = Bytes.toBytes("name");
        MultipleColumnPrefixFilter mcpf = new MultipleColumnPrefixFilter(prefixes);
        list.addFilter(mcpf);
        scan.setFilter(list);

        ResultScanner rs = hTable.getScanner(scan);
        Iterator<Result> iter = rs.iterator();
        while (iter.hasNext()){
            Result result = iter.next();
            printResult(result);
        }
    }

    /*
    ??Result??
     */
    static void printResult(Result result){
        System.out.println("***********"+Bytes.toString(result.getRow()));
        NavigableMap<byte[],NavigableMap<byte[], NavigableMap<Long,byte[]>>> map = result.getMap();
        for(Map.Entry<byte[],NavigableMap<byte[],NavigableMap<Long,byte[]>>> entry: map.entrySet()){
            String family = Bytes.toString(entry.getKey());
            for(Map.Entry<byte[],NavigableMap<Long,byte[]>> columnEntry :entry.getValue().entrySet()){
                String column = Bytes.toString(columnEntry.getKey());
                String value = "";
                if("age".equals(column)){
                    value=""+Bytes.toInt(columnEntry.getValue().firstEntry().getValue());
                }else {
                    value=""+Bytes.toString(columnEntry.getValue().firstEntry().getValue());
                }
                System.out.println(family+":"+column+":"+value);
            }
        }
    }
}